import { Inject, Injectable } from "@nestjs/common";
import kafka from 'kafka-node'
import { MongoService } from "./mongo.service";

import { Kafka, logLevel, Producer } from 'kafkajs'
import { MsgService } from "./msg.service";
import { RedisService } from "./redis.service";

@Injectable()
export class TransferKafkaService {
    client: kafka.KafkaClient = null;
    //consumer: kafka.Consumer = null;
    producerMongo: Producer = null;
    producerPush: Producer = null;

    kafka: Kafka
    kafkaHost: string = '134.175.224.228:9092'

    constructor(
        private mongoService: MongoService,
        private msgService : MsgService,
        private redisService : RedisService
    ) { }

    async onModuleInit() {
        this.kafka = new Kafka({
            logLevel: logLevel.INFO,
            brokers: [`${this.kafkaHost}`],
            clientId: 'kafka-transfer',
        })

        const topic = 'latestMsgToRedis'
        const consumer = this.kafka.consumer({ groupId: 'transfer' })

        this.producerMongo = this.kafka.producer({
            allowAutoTopicCreation: true,
            transactionTimeout: 30000
        })

        this.producerPush = this.kafka.producer({
            allowAutoTopicCreation: true,
            transactionTimeout: 30000
        })

        await Promise.all([
            this.producerMongo.connect(),
            this.producerPush.connect()
        ])

        const run = async () => {
            await consumer.connect()
            await consumer.subscribe({ topic, fromBeginning: true })
            await consumer.run({
                eachMessage: async ({ topic, partition, message }) => {

                    let msg: any = JSON.parse(message.value.toString())

                    if(msg.TypeName != 'AddMsg')
                        return

                    console.log(`============offset:=>${message.offset}`)

                    let msgJson = JSON.stringify(msg)

                    await Promise.all([
                        //发送到推送队列
                        this.producerPush.send({
                            topic: 'MsgToPush',
                            messages: [{
                                key: Date.now().toString(),
                                value: msgJson,
                                partition: 0
                            }
                            ]
                        }),

                        //发送到存储队列{
                        this.producerMongo.send({
                            topic: 'MsgToMongo',
                            messages: [{
                                key: Date.now().toString(),
                                value: msgJson,
                                partition: 0
                            }]
                        })
                    ])


                },
            })
        }

        run().catch(e =>
            console.error(`[example/consumer] ${e.message}`, e)
        )
    }

    async onModuleDestroy() {

    }
}